package com.burges.net.dataSet.iterator

import org.apache.flink.api.scala.{DataSet, ExecutionEnvironment}

/**
  * 增量迭代计算代码示例
  */
object DeltaIterations {
  def main(args: Array[String]): Unit = {
    val env = ExecutionEnvironment.getExecutionEnvironment
    // read the initial data sets
    // 读取初始化数据集
//    val initialSolutionSet: DataSet[(Long, Double)] = // {...}
    // 读取初始化WorkSet数据集
//    val initialWorkset: DataSet[(Long, Double)] = // [...]

    // 设定迭代参数
//    val maxIterations = 100
//    val keyPosition = 0
//
    // 通过iterateDelta应用增量迭代方法
//    val result = initialSolutionSet.iterateDelta(initialWorkset, maxIterations, Array(keyPosition)) {
//      (solution, workset) =>
//        val candidateUpdates = workset.groupBy(1).reduceGroup(new ComputeCandidateChanges())
//        val deltas = candidateUpdates.join(solution).where(0).equalTo(0)(new CompareChangesToCurrent())
//
//        val nextWorkset = deltas.filter(new FilterByThreshold())
//
//        (deltas, nextWorkset)
//    }
//
    // 输出迭代计算的结果
//    result.writeAsCsv(outputPath)

    env.execute()
  }

}
